package com.ds.lens.flink.monitor.monitorjob.function;

import com.ds.lens.data.common.document.dto.MonitorEvent;
import com.ds.lens.data.common.document.dynamic.DynamicDetectionResult;
import com.ds.lens.data.common.document.dynamic.DynamicMetricBO;
import com.ds.lens.data.common.document.dynamic.DynamicSinkBO;
import com.ds.lens.flink.monitor.common.BuilderUtil;
import com.ds.lens.flink.monitor.common.bo.MetricBundle;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Map;

/**
 * Description:
 *
 * @author WeiShaoying
 * @date 2020-05-14
 */
public class DynamicDetectionResultFunction extends ProcessFunction<Tuple2<MetricBundle, Map<String, MonitorEvent>>, DynamicDetectionResult> {
    private static final long serialVersionUID = 126989212243826910L;

    @Override
    public void processElement(Tuple2<MetricBundle, Map<String, MonitorEvent>> tuple2, Context context, Collector<DynamicDetectionResult> collector) throws Exception {
        Map<String, MonitorEvent> monitorEventMap = tuple2.f1;
        if (CollectionUtils.isEmpty(monitorEventMap)) {
            return;
        }
        monitorEventMap.forEach((key, monitorEvent) -> {
            DynamicSinkBO dynamicSinkBO = monitorEvent.getDynamicSinkBO();
            if (dynamicSinkBO != null && !CollectionUtils.isEmpty(dynamicSinkBO.getMetricBOS())) {
                dynamicSinkBO.getMetricBOS().sort((o1, o2) -> {
                    if (o1.getTimestamp() < o2.getTimestamp()) {
                        return -1;
                    } else if (o1.getTimestamp() > o2.getTimestamp()) {
                        return 1;
                    } else {
                        return 0;
                    }
                });
                List<DynamicMetricBO> bos = dynamicSinkBO.getMetricBOS();
                DynamicMetricBO bo = bos.get(bos.size() - 1);
                if (bo.getTimestamp().equals(monitorEvent.getTriggered().getEndTime() - monitorEvent.getInterval() * 60000)) {
                    DynamicDetectionResult detectionResult = new DynamicDetectionResult();
                    detectionResult.setMonitorId(monitorEvent.getMonitorKey().getMonitorId());
                    detectionResult.setVersion(monitorEvent.getMonitorKey().getVersion());
                    detectionResult.setForecast(bo.getForecast());
                    detectionResult.setActual(bo.getActual());
                    detectionResult.setIsAnomaly(bo.getIsAnomaly());
                    detectionResult.setMetricBy(key);
                    String subTopic = BuilderUtil.generateKey(detectionResult.getMonitorId(), detectionResult.getVersion(), monitorEvent.getMetricName(), monitorEvent.getAggrEnum(), monitorEvent.getGroupBys());
                    detectionResult.setSubTopic(subTopic);
                    detectionResult.setTimestamp(bo.getTimestamp());
                    detectionResult.setAlertThresholdUpper(bo.getAlertThresholdUpper());
                    detectionResult.setAlertThresholdLower(bo.getAlertThresholdLower());
                    collector.collect(detectionResult);
                }
            }
        });

    }
}
